﻿/*
-----------------------------------------------------------------------------
This source file is part of Cell Cloud.

Copyright (c) 2009-2012 Cell Cloud Team (cellcloudproject@gmail.com)

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.
-----------------------------------------------------------------------------
*/

#include "CellStableHeaders.h"
#include "CellFileExpressTask.h"
#include "CellFileExpressContext.h"
#include "CellFileExpressListener.h"
#include "CellFileExpressDefinition.h"
#include "CellFileAttribute.h"
#include "CellThread.h"
#include "CellBlockingConnector.h"
#include "CellSession.h"
#include "CellPacket.h"
#include "CellMessage.h"
#include "CellSingleFileStorage.h"
#include "CellSingleFileResultSet.h"
#include "CellStringConverter.h"
#include "CellLoggerManager.h"

namespace cell
{
//---------------------------------------------------------------------------
FileExpressTask::FileExpressTask(FileExpressContext* context)
	: mContext(context)
	, mConnector(NULL)
	, mListener(NULL)
	, mProgress(0)
	, mDataCache(NULL)
	, mDataLength(0)
	, mState(EXPRESS_STATE_UNKNOWN)
	, mFileStorage(NULL)
{
	mReceiveCache = CELL_ALLOC_T(char, FE_CACHE_SIZE, MEMCATEGORY_GENERAL);
	memset(mReceiveCache, 0x0, FE_CACHE_SIZE);
}
//---------------------------------------------------------------------------
FileExpressTask::~FileExpressTask(void)
{
	CELL_FREE(mReceiveCache, MEMCATEGORY_GENERAL);

	if (NULL != mFileStorage)
	{
		mFileStorage->close();

		CELL_DELETE mFileStorage;
		mFileStorage = NULL;
	}
}
//---------------------------------------------------------------------------
void FileExpressTask::run(void)
{
	switch (mContext->getOperate())
	{
	case FileExpressContext::OP_UPLOAD:
		upload();
		break;
	case FileExpressContext::OP_DOWNLOAD:
		download();
		break;
	default:
		char log[256] = {0x0};
		sprintf(log, "[CC] File express task error - file: '%s'", mContext->getFileName().c_str());
		LoggerManager::getSingleton().log(log);
		break;
	}
}
//---------------------------------------------------------------------------
void FileExpressTask::abort(void)
{
	CELL_LOCK_MUTEX(mMonitor);

	mState = EXPRESS_STATE_EXIT;

	CELL_THREAD_NOTIFY_ONE(mWait);
}
//---------------------------------------------------------------------------
FileExpressContext* FileExpressTask::getContext(void) const
{
	return mContext;
}
//---------------------------------------------------------------------------
void FileExpressTask::setListener(FileExpressListener* listener)
{
	mListener = listener;
}
//---------------------------------------------------------------------------
void FileExpressTask::sessionCreated(Session* session)
{
}
//---------------------------------------------------------------------------
void FileExpressTask::sessionDestroyed(Session* session)
{
	CELL_LOCK_MUTEX(mMonitor);

	mState = EXPRESS_STATE_LOST;

	CELL_THREAD_NOTIFY_ONE(mWait);
}
//---------------------------------------------------------------------------
void FileExpressTask::sessionOpened(Session* session)
{
}
//---------------------------------------------------------------------------
void FileExpressTask::sessionClosed(Session* session)
{
}
//---------------------------------------------------------------------------
void FileExpressTask::messageReceived(Session* session, Message* message)
{
	memset(mReceiveCache, 0x0, FE_CACHE_SIZE);
	uint32 len = message->get(mReceiveCache);

	Packet packet;
	if (Packet::unpack(&packet, mReceiveCache, len))
	{
		interpret(session, &packet);
	}
	else
	{
		mContext->setErrorCode(FileExpressContext::EC_PACKET_ERROR);
		if (NULL != mListener)
		{
			mListener->expressError(mContext);
		}
	}
}
//---------------------------------------------------------------------------
void FileExpressTask::messageSent(Session* session, Message* message)
{
}
//---------------------------------------------------------------------------
void FileExpressTask::errorOccurred(uint16 errorCode, Session* session)
{
	// 设置错误码
	mContext->setErrorCode(FileExpressContext::EC_NETWORK_FAULT);
	if (NULL != mListener)
	{
		mListener->expressError(mContext);
	}

	CELL_LOCK_MUTEX(mMonitor);

	mState = EXPRESS_STATE_EXIT;

	CELL_THREAD_NOTIFY_ONE(mWait);
}
//---------------------------------------------------------------------------
void FileExpressTask::interpret(Session* session, Packet* packet)
{
	char tag[5] = {0x0};
	packet->getTag(tag);

	CELL_LOCK_MUTEX(mMonitor);

	if (tag[0] == FEPT_DATA_RECEIPT_B1 && tag[1] == FEPT_DATA_RECEIPT_B2
		&& tag[2] == FEPT_DATA_RECEIPT_B3 && tag[3] == FEPT_DATA_RECEIPT_B4)
	{
		// 包格式：文件名|数据进度

		if (packet->getSubsegmentNum() < 2)
		{
			mContext->setErrorCode(FileExpressContext::EC_PACKET_ERROR);
			if (NULL != mListener)
			{
				mListener->expressError(mContext);
			}

			mState = EXPRESS_STATE_EXIT;
			CELL_THREAD_NOTIFY_ONE(mWait);
			return;
		}

		mState = EXPRESS_STATE_DATA;

		CELL_THREAD_NOTIFY_ONE(mWait);
	}
	else if (tag[0] == FEPT_DATA_B1 && tag[1] == FEPT_DATA_B2
		&& tag[2] == FEPT_DATA_B3 && tag[3] == FEPT_DATA_B4)
	{
		// 包格式：授权码|文件名|数据起始位|数据结束位|数据

		if (packet->getSubsegmentNum() < 5)
		{
			mContext->setErrorCode(FileExpressContext::EC_PACKET_ERROR);
			if (NULL != mListener)
			{
				mListener->expressError(mContext);
			}

			mState = EXPRESS_STATE_EXIT;
			CELL_THREAD_NOTIFY_ONE(mWait);
			return;
		}

		char szStart[16] = {0x0};
		packet->getSubsegment(szStart, 2);
		char szEnd[16] = {0x0};
		packet->getSubsegment(szEnd, 3);
		uint start = StringConverter::parseUnsignedInt(szStart);
		uint end = StringConverter::parseUnsignedInt(szEnd);

		memset(mDataCache, 0x0, FE_CACHE_SIZE);
		mDataLength = packet->getSubsegment(mDataCache, 4);

		if (end - start == mDataLength)
		{
			mProgress = start;

			mState = EXPRESS_STATE_DATA;
		}
		else
		{
			mContext->setErrorCode(FileExpressContext::EC_PACKET_ERROR);
			if (NULL != mListener)
			{
				mListener->expressError(mContext);
			}

			mState = EXPRESS_STATE_EXIT;
		}

		CELL_THREAD_NOTIFY_ONE(mWait);
	}
	else if (tag[0] == FEPT_END_B1 && tag[1] == FEPT_END_B2
		&& tag[2] == FEPT_END_B3 && tag[3] == FEPT_END_B4)
	{
		// 包格式：文件名|文件长度

		if (packet->getSubsegmentNum() < 2)
		{
			mContext->setErrorCode(FileExpressContext::EC_PACKET_ERROR);
			if (NULL != mListener)
			{
				mListener->expressError(mContext);
			}

			mState = EXPRESS_STATE_EXIT;
			CELL_THREAD_NOTIFY_ONE(mWait);
			return;
		}

		mState = EXPRESS_STATE_END;

		CELL_THREAD_NOTIFY_ONE(mWait);
	}
	else if (tag[0] == FEPT_BEGIN_B1 && tag[1] == FEPT_BEGIN_B2
		&& tag[2] == FEPT_BEGIN_B3 && tag[3] == FEPT_BEGIN_B4)
	{
		// 包格式：文件名|文件长度

		if (packet->getSubsegmentNum() < 2)
		{
			mContext->setErrorCode(FileExpressContext::EC_PACKET_ERROR);
			if (NULL != mListener)
			{
				mListener->expressError(mContext);
			}

			mState = EXPRESS_STATE_EXIT;
			CELL_THREAD_NOTIFY_ONE(mWait);
			return;
		}

		mState = EXPRESS_STATE_BEGIN;

		CELL_THREAD_NOTIFY_ONE(mWait);
	}
	else if (tag[0] == FEPT_ATTR_B1 && tag[1] == FEPT_ATTR_B2
		&& tag[2] == FEPT_ATTR_B3 && tag[3] == FEPT_ATTR_B4)
	{
		// 包格式：文件名|属性序列

		if (packet->getSubsegmentNum() < 2)
		{
			mContext->setErrorCode(FileExpressContext::EC_PACKET_ERROR);
			if (NULL != mListener)
			{
				mListener->expressError(mContext);
			}

			mState = EXPRESS_STATE_EXIT;
			CELL_THREAD_NOTIFY_ONE(mWait);
			return;
		}

		// 文件名
		char file[256] = {0x0};
		packet->getSubsegment(file, 0);

		if (mContext->getFileName() == file)
		{
			// 还原服务器端文件属性
			char attrstr[1024] = {0x0};
			uint len = packet->getSubsegment(attrstr, 1);
			mContext->getAttribute()->deserialize(attrstr, len);

			mState = EXPRESS_STATE_PREPARE;
		}
		else
		{
			mContext->setErrorCode(FileExpressContext::EC_PACKET_ERROR);
			if (NULL != mListener)
			{
				mListener->expressError(mContext);
			}

			mState = EXPRESS_STATE_EXIT;
		}

		CELL_THREAD_NOTIFY_ONE(mWait);
	}
	else if (tag[0] == FEPT_AUTH_B1 && tag[1] == FEPT_AUTH_B2
		&& tag[2] == FEPT_AUTH_B3 && tag[3] == FEPT_AUTH_B4)
	{
		// 权限判定
		char auth[FEA_DML] = {0x0};
		packet->getSubsegment(auth, 0);
		mContext->getAuthCode()->changeAuth(auth);

		bool authorized = false;

		switch (mContext->getOperate())
		{
		case FileExpressContext::OP_UPLOAD:
			if (mContext->getAuthCode()->getAuth() == ExpressAuthCode::WRITE)
			{
				authorized = true;
			}
			break;
		case FileExpressContext::OP_DOWNLOAD:
			if (mContext->getAuthCode()->getAuth() == ExpressAuthCode::WRITE
				|| mContext->getAuthCode()->getAuth() == ExpressAuthCode::READ)
			{
				authorized = true;
			}
			break;
		default:
			break;
		}

		if (authorized)
			mState = EXPRESS_STATE_ATTR;
		else
			mState = EXPRESS_STATE_UNAUTH;

		CELL_THREAD_NOTIFY_ONE(mWait);
	}
	else
	{
		mContext->setErrorCode(FileExpressContext::EC_PACKET_ERROR);
		if (NULL != mListener)
		{
			mListener->expressError(mContext);
		}

		mState = EXPRESS_STATE_EXIT;
		CELL_THREAD_NOTIFY_ONE(mWait);
	}
}
//---------------------------------------------------------------------------
void FileExpressTask::upload(void)
{
	ResultSet* resultSet = NULL;

	if (NULL == mFileStorage)
	{
		mFileStorage = CELL_NEW SingleFileStorage("FileExpress");

		// 打开存储器。
		mFileStorage->open(mContext->getFullPath());
	}

	// 由存储器创建结果集
	resultSet = mFileStorage->createResultSet();

	// 读取文件
	mFileStorage->store(resultSet, mFileStorage->getReadStatement());
	// 移动游标到数据位
	resultSet->next();

	// 检查文件是否存在
	if (!resultSet->getBool(SingleFileStorage::FIELD_EXIST))
	{
		// 文件不存在
		mFileStorage->destroyResultSet(resultSet);
		mFileStorage->close();

		CELL_DELETE mFileStorage;
		mFileStorage = NULL;

	#ifdef CELL_DEBUG_MODE
		char log[512] = {0x0};
		sprintf(log, "[CC] Task detected the file '%s' does not exist.", mContext->getFullPath().c_str());
		LoggerManager::getSingleton().log(log);
	#endif // CELL_DEBUG_MODE

		// 设置错误码
		mContext->setErrorCode(FileExpressContext::EC_FILE_NOEXIST);
		if (NULL != mListener)
		{
			// 通知监听器发生错误
			mListener->expressError(mContext);
		}

		return;
	}

	if (NULL == mConnector)
	{
		mConnector = CELL_NEW BlockingConnector();
		char head[4] = {0x10, 0x04, 0x11, 0x24};
		char tail[4] = {0x11, 0x24, 0x10, 0x04};
		mConnector->defineDataMark(head, 4, tail, 4);
		mConnector->setConnectTimeout(5000);
		mConnector->setHandler(this);
	}

	Session* session = NULL;

	// 设置初始状态
	mState = EXPRESS_STATE_LOST;

	while (mState != EXPRESS_STATE_EXIT)
	{
		switch (mState)
		{
		case EXPRESS_STATE_LOST:
		{
			if (false == mConnector->isConnected())
			{
				// 连接
				session = mConnector->connect(*mContext->getAddress());
				if (NULL == session)
				{
					// 设置错误码
					mContext->setErrorCode(FileExpressContext::EC_NETWORK_FAULT);
					if (NULL != mListener)
					{
						// 通知监听器发生错误
						mListener->expressError(mContext);
					}

					// 5 秒后重试
					Thread::sleep(5000);
					break;
				}
			}

			CELL_LOCK_MUTEX_NAMED(mMonitor, lock);

			// 发送验证码进行验证。
			postAuthCode();

			CELL_THREAD_WAIT(mWait, mutex, lock);

			break;
		}

		case EXPRESS_STATE_ATTR:
		{
			CELL_LOCK_MUTEX_NAMED(mMonitor, lock);

			// 请求文件报告
			queryFile();

			CELL_THREAD_WAIT(mWait, mutex, lock);

			break;
		}

		case EXPRESS_STATE_PREPARE:
		{
			CELL_LOCK_MUTEX_NAMED(mMonitor, lock);

			if (!mContext->getAttribute()->exist())
			{
				// 设置文件属性
				FileAttribute* attr = mContext->getAttribute();
				attr->size(resultSet->getLong(SingleFileStorage::FIELD_SIZE));

				// 新上传
				mProgress = 0;

				// 准备开始
				beginUpload(resultSet);
			}
			else
			{
				// 文件存在，判断是否续传
				// 判断文件大小
				long remoteFileSize = mContext->getAttribute()->size();
				long localFileSize = resultSet->getLong(SingleFileStorage::FIELD_SIZE);
				if (remoteFileSize == localFileSize)
				{
					mState = EXPRESS_STATE_EXIT;

					// 设置错误码
					mContext->setErrorCode(FileExpressContext::EC_REJECT_SIZE);
					if (NULL != mListener)
					{
						// 通知监听器发生错误
						mListener->expressError(mContext);
					}

					break;
				}
				else
				{
					// 如果服务端文件小于本地文件则续传
					if (remoteFileSize < localFileSize)
					{
						mProgress = remoteFileSize;

						// 准备开始
						beginUpload(resultSet);
					}
					else
					{
						mState = EXPRESS_STATE_EXIT;

						// 设置错误码
						mContext->setErrorCode(FileExpressContext::EC_REJECT_SIZE);
						if (NULL != mListener)
						{
							// 通知监听器发生错误
							mListener->expressError(mContext);
						}

						break;
					}
				}
			}

			CELL_THREAD_WAIT(mWait, mutex, lock);

			break;
		}

		case EXPRESS_STATE_BEGIN:
		{
			CELL_LOCK_MUTEX_NAMED(mMonitor, lock);

			processUploadData(resultSet);

			CELL_THREAD_WAIT(mWait, mutex, lock);

			break;
		}

		case EXPRESS_STATE_DATA:
		{
			CELL_LOCK_MUTEX_NAMED(mMonitor, lock);

			if (false == processUploadData(resultSet))
			{
				endUpload(resultSet);
			}

			CELL_THREAD_WAIT(mWait, mutex, lock);

			break;
		}

		case EXPRESS_STATE_END:
		{
			CELL_LOCK_MUTEX(mMonitor);

			// 结束任务循环
			mState = EXPRESS_STATE_EXIT;

			break;
		}

		case EXPRESS_STATE_UNAUTH:
		{
			char log[512] = {0x0};
			sprintf(log, "Upload file '%s' have not been authorized", mContext->getFullPath().c_str());
			LoggerManager::getSingleton().log(log);

			// 结束任务循环
			mState = EXPRESS_STATE_EXIT;

			// 设置错误码
			mContext->setErrorCode(FileExpressContext::EC_UNAUTH);
			if (NULL != mListener)
			{
				// 通知监听器发生错误
				mListener->expressError(mContext);
			}

			break;
		}

		default:
			break;
		}
	}

	// 关闭连接
	if (NULL != mConnector)
	{
		// 让出 CPU
		Thread::sleep(100);

		mConnector->disconnect();
		CELL_DELETE mConnector;
		mConnector = NULL;
	}

	if (NULL != mFileStorage)
	{
		mFileStorage->destroyResultSet(resultSet);
		mFileStorage->close();

		CELL_DELETE mFileStorage;
		mFileStorage = NULL;
	}
}
//---------------------------------------------------------------------------
void FileExpressTask::download(void)
{
	if (NULL == mDataCache)
	{
		mDataCache = CELL_ALLOC_T(char, FE_CACHE_SIZE, MEMCATEGORY_GENERAL);
	}

	ResultSet* resultSet = NULL;

	if (NULL == mFileStorage)
	{
		mFileStorage = CELL_NEW SingleFileStorage("FileExpress");

		// 打开存储器。
		mFileStorage->open(mContext->getFullPath());
	}

	// 由存储器创建结果集
	resultSet = mFileStorage->createResultSet();

	// 读文件
	mFileStorage->store(resultSet, mFileStorage->getReadStatement());
	// 移动游标到数据位
	resultSet->next();

	mProgress = 0;

	// 检查文件是否存在
	if (resultSet->getBool(SingleFileStorage::FIELD_EXIST))
	{
		// 文件已经存在，设置偏移以续传
		mProgress = resultSet->getLong(SingleFileStorage::FIELD_SIZE);
	}

	// 销毁结果集
	mFileStorage->destroyResultSet(resultSet);

	// 写文件结果集
	resultSet = mFileStorage->createResultSet();
	mFileStorage->store(resultSet, mFileStorage->getWriteStatement());
	resultSet->next();

	if (NULL == mConnector)
	{
		mConnector = CELL_NEW BlockingConnector();
		char head[4] = {0x10, 0x04, 0x11, 0x24};
		char tail[4] = {0x11, 0x24, 0x10, 0x04};
		mConnector->defineDataMark(head, 4, tail, 4);
		mConnector->setConnectTimeout(5000);
		mConnector->setHandler(this);
	}

	// 设置初始状态
	mState = EXPRESS_STATE_LOST;

	while (mState != EXPRESS_STATE_EXIT)
	{
		switch (mState)
		{
		case EXPRESS_STATE_LOST:
		{
			if (false == mConnector->isConnected())
			{
				// 连接
				Session* session = mConnector->connect(*mContext->getAddress());
				if (NULL == session)
				{
					// 设置错误码
					mContext->setErrorCode(FileExpressContext::EC_NETWORK_FAULT);
					if (NULL != mListener)
					{
						// 通知监听器发生错误
						mListener->expressError(mContext);
					}

					// 5 秒后重试
					Thread::sleep(5000);
					break;
				}
			}

			CELL_LOCK_MUTEX_NAMED(mMonitor, lock);

			// 发送验证码进行验证。
			postAuthCode();

			CELL_THREAD_WAIT(mWait, mutex, lock);

			break;
		}

		case EXPRESS_STATE_ATTR:
		{
			CELL_LOCK_MUTEX_NAMED(mMonitor, lock);

			// 请求文件报告
			queryFile();

			CELL_THREAD_WAIT(mWait, mutex, lock);

			break;
		}

		case EXPRESS_STATE_PREPARE:
		{
			CELL_LOCK_MUTEX_NAMED(mMonitor, lock);

			// 判断文件是否需要下载
			FileAttribute* attr = mContext->getAttribute();
			if (!attr->exist() || attr->size() <= mProgress)
			{
				// 服务器端文件不存在，不进行下载
				// 服务器端文件小于或者等于本地文件大小，不进行下载
				mState = EXPRESS_STATE_EXIT;

				// 设置错误码
				mContext->setErrorCode((attr->exist()) ? FileExpressContext::EC_REJECT_SIZE : FileExpressContext::EC_FILE_NOEXIST);
				if (NULL != mListener)
				{
					// 通知监听器发生错误
					mListener->expressError(mContext);
				}
			}
			else
			{
				beginDownload();

				CELL_THREAD_WAIT(mWait, mutex, lock);
			}

			break;
		}

		case EXPRESS_STATE_BEGIN:
		{
			CELL_LOCK_MUTEX_NAMED(mMonitor, lock);

			offerDownload(resultSet);

			CELL_THREAD_WAIT(mWait, mutex, lock);

			break;
		}

		case EXPRESS_STATE_DATA:
		{
			CELL_LOCK_MUTEX_NAMED(mMonitor, lock);

			processDownloadData(resultSet);

			CELL_THREAD_WAIT(mWait, mutex, lock);

			break;
		}

		case EXPRESS_STATE_END:
		{
			CELL_LOCK_MUTEX(mMonitor);

			endDownload(resultSet);

			// 结束任务循环
			mState = EXPRESS_STATE_EXIT;

			break;
		}

		case EXPRESS_STATE_UNAUTH:
		{
			char log[512] = {0x0};
			sprintf(log, "Download file '%s' have not been authorized", mContext->getFullPath().c_str());
			LoggerManager::getSingleton().log(log);

			// 结束任务循环
			mState = EXPRESS_STATE_EXIT;

			// 设置错误码
			mContext->setErrorCode(FileExpressContext::EC_UNAUTH);
			if (NULL != mListener)
			{
				// 通知监听器发生错误
				mListener->expressError(mContext);
			}

			break;
		}

		default:
			break;
		}
	}

	// 关闭连接
	if (NULL != mConnector)
	{
		// 让出 CPU
		Thread::sleep(100);

		mConnector->disconnect();
		CELL_DELETE mConnector;
		mConnector = NULL;
	}

	if (NULL != mFileStorage)
	{
		mFileStorage->destroyResultSet(resultSet);
		mFileStorage->close();

		CELL_DELETE mFileStorage;
		mFileStorage = NULL;
	}

	if (NULL != mDataCache)
	{
		CELL_FREE(mDataCache, MEMCATEGORY_GENERAL);
		mDataCache = NULL;
	}
}
//---------------------------------------------------------------------------
void FileExpressTask::postAuthCode(void)
{
	// 包格式：授权码

	char ptg[5] = {0x0};
	FEPT_AUTH_ASSIGN(ptg);
	Packet packet(ptg, 1);

	String authCode = mContext->getAuthCode()->getCode();
	packet.appendSubsegment(authCode.c_str(), authCode.length());

	char data[128] = {0x0};
	uint len = Packet::pack(data, &packet);
	Message message(data, len);
	mConnector->write(&message);
}
//---------------------------------------------------------------------------
void FileExpressTask::queryFile(void)
{
	// 包格式：授权码|文件名

	char ptg[5] = {0x0};
	FEPT_ATTR_ASSIGN(ptg);
	Packet packet(ptg, 2);

	String authCode = mContext->getAuthCode()->getCode();
	packet.appendSubsegment(authCode.c_str(), authCode.length());

	String fileName = mContext->getFileName();
	packet.appendSubsegment(fileName.c_str(), fileName.length());

	char data[512] = {0x0};
	uint len = Packet::pack(data, &packet);
	Message message(data, len);
	mConnector->write(&message);
}
//---------------------------------------------------------------------------
void FileExpressTask::beginUpload(ResultSet* resultSet)
{
	String authCode = mContext->getAuthCode()->getCode();

	String filename = mContext->getFileName();
	long filesize = resultSet->getLong(SingleFileStorage::FIELD_SIZE);
	String szSize = StringConverter::toString(filesize);

	String operate = StringConverter::toString(FileExpressContext::OP_UPLOAD);

	// 包格式：授权码|文件名|文件长度|操作

	char ptag[5] = {0x0};
	FEPT_BEGIN_ASSIGN(ptag);

	Packet packet(ptag, 3);
	packet.appendSubsegment(authCode.c_str(), authCode.length());
	packet.appendSubsegment(filename.c_str(), filename.length());
	packet.appendSubsegment(szSize.c_str(), szSize.length());
	packet.appendSubsegment(operate.c_str(), operate.length());

	char data[1024] = {0x0};
	uint len = Packet::pack(data, &packet);
	Message message(data, len);
	mConnector->write(&message);

	// 设置总字节数
	mContext->setBytesTotal(filesize - mProgress);

	// 通知监听器开始
	if (NULL != mListener)
	{
		mListener->expressStarted(mContext);
	}
}
//---------------------------------------------------------------------------
void FileExpressTask::endUpload(ResultSet* resultSet)
{
	String authCode = mContext->getAuthCode()->getCode();

	String filename = mContext->getFileName();
	long filesize = resultSet->getLong(SingleFileStorage::FIELD_SIZE);
	String szSize = StringConverter::toString(filesize);

	String operate = StringConverter::toString(FileExpressContext::OP_UPLOAD);

	// 包结构：授权码|文件名|文件长度|操作

	// 文件数据发送完毕
	char ptag[5] = {0x0};
	FEPT_END_ASSIGN(ptag);

	Packet packet(ptag, 5);
	packet.appendSubsegment(authCode.c_str(), authCode.length());
	packet.appendSubsegment(filename.c_str(), filename.length());
	packet.appendSubsegment(szSize.c_str(), szSize.length());
	packet.appendSubsegment(operate.c_str(), operate.length());

	char data[1024] = {0x0};
	uint len = Packet::pack(data, &packet);
	Message message(data, len);
	mConnector->write(&message);

	// 通知监听器结束
	if (NULL != mListener)
	{
		mListener->expressCompleted(mContext);
	}
}
//---------------------------------------------------------------------------
bool FileExpressTask::processUploadData(ResultSet* resultSet)
{
	// 包格式：授权码|文件名|数据起始位|数据结束位|数据

	int length = 0;
	char buf[FE_FILEDATA_SIZE] = {0x0};
	if ((length = resultSet->getRaw(SingleFileStorage::FIELD_DATA, buf, mProgress, FE_FILEDATA_SIZE)) > 0)
	{
		char ptg[5] = {0x0};
		FEPT_DATA_ASSIGN(ptg);

		char pdata[FE_CACHE_SIZE] = {0x0};
		uint plen = 0;

		Packet packet(ptg, 4);

		String authCode = mContext->getAuthCode()->getCode();
		String filename = mContext->getFileName();
		packet.appendSubsegment(authCode.c_str(), authCode.length());
		packet.appendSubsegment(filename.c_str(), filename.length());

		String start = StringConverter::toString(mProgress);
		String end = StringConverter::toString(mProgress + length);
		packet.appendSubsegment(start.c_str(), start.length());
		packet.appendSubsegment(end.c_str(), end.length());

		// 数据
		packet.appendSubsegment(buf, length);

		plen = Packet::pack(pdata, &packet);
		Message message(pdata, plen);
		mConnector->write(&message);

		// 更新进度
		mProgress += length;

		// 设置已传输大小
		mContext->setBytesLoaded(mProgress);

		// 通知监听器正在进行数据传输
		if (NULL != mListener)
		{
			mListener->expressProgress(mContext);
		}

		return true;
	}
	else
	{
		return false;
	}
}
//---------------------------------------------------------------------------
void FileExpressTask::beginDownload()
{
	String authCode = mContext->getAuthCode()->getCode();

	String filename = mContext->getFileName();
	long filesize = mContext->getAttribute()->size();
	String szSize = StringConverter::toString(filesize);

	String operate = StringConverter::toString(FileExpressContext::OP_DOWNLOAD);

	// 包格式：授权码|文件名|文件长度|操作

	char ptag[5] = {0x0};
	FEPT_BEGIN_ASSIGN(ptag);

	Packet packet(ptag, 3);
	packet.appendSubsegment(authCode.c_str(), authCode.length());
	packet.appendSubsegment(filename.c_str(), filename.length());
	packet.appendSubsegment(szSize.c_str(), szSize.length());
	packet.appendSubsegment(operate.c_str(), operate.length());

	char data[1024] = {0x0};
	uint len = Packet::pack(data, &packet);
	Message message(data, len);
	mConnector->write(&message);

	// 设置总字节数
	mContext->setBytesTotal(filesize - mProgress);

	// 通知监听器开始
	if (NULL != mListener)
	{
		mListener->expressStarted(mContext);
	}
}
//---------------------------------------------------------------------------
void FileExpressTask::endDownload(ResultSet* resultSet)
{
	if (NULL != mFileStorage)
	{
		mFileStorage->close();
	}

	String authCode = mContext->getAuthCode()->getCode();

	String filename = mContext->getFileName();
	long filesize = mContext->getAttribute()->size();
	String szSize = StringConverter::toString(filesize);

	String operate = StringConverter::toString(FileExpressContext::OP_DOWNLOAD);

	// 包结构：授权码|文件名|文件长度|操作

	char ptag[5] = {0x0};
	FEPT_END_ASSIGN(ptag);

	Packet packet(ptag, 6);
	packet.appendSubsegment(authCode.c_str(), authCode.length());
	packet.appendSubsegment(filename.c_str(), filename.length());
	packet.appendSubsegment(szSize.c_str(), szSize.length());
	packet.appendSubsegment(operate.c_str(), operate.length());

	char data[1024] = {0x0};
	uint len = Packet::pack(data, &packet);
	Message message(data, len);
	mConnector->write(&message);

	// 通知监听器开始
	if (NULL != mListener)
	{
		mListener->expressCompleted(mContext);
	}
}
//---------------------------------------------------------------------------
void FileExpressTask::offerDownload(ResultSet* resultSet)
{
	// 设置文件长度
	resultSet->updateLong(SingleFileStorage::FIELD_SIZE, mContext->getAttribute()->size());

	// 包格式：授权码|文件名|文件操作起始位置

	String authCode = mContext->getAuthCode()->getCode();
	String filename = mContext->getFileName();
	char offset[32] = {0x0};
	sprintf(offset, "%ld", mProgress);

	char ptag[5] = {0x0};
	FEPT_OFFER_ASSIGN(ptag);

	Packet packet(ptag, 4);
	packet.appendSubsegment(authCode.c_str(), authCode.length());
	packet.appendSubsegment(filename.c_str(), filename.length());
	packet.appendSubsegment(offset, strlen(offset));

	char data[1024] = {0x0};
	uint len = Packet::pack(data, &packet);
	Message message(data, len);
	mConnector->write(&message);
}
//---------------------------------------------------------------------------
void FileExpressTask::processDownloadData(ResultSet* resultSet)
{
	// 写数据
	resultSet->updateRaw(SingleFileStorage::FIELD_DATA, mDataCache, mProgress, mDataLength);

	// 更新进度
	mProgress += mDataLength;

	// 发送数据回执

	// 包格式：授权码|文件名|新数据进度

	String authCode = mContext->getAuthCode()->getCode();
	String filename = mContext->getFileName();
	char szProgress[16] = {0x0};
	sprintf(szProgress, "%ld", mProgress);

	char ptag[5] = {0x0};
	FEPT_DATA_RECEIPT_ASSIGN(ptag);

	Packet packet(ptag, 5);
	packet.appendSubsegment(authCode.c_str(), authCode.length());
	packet.appendSubsegment(filename.c_str(), filename.length());
	packet.appendSubsegment(szProgress, strlen(szProgress));

	char data[1024] = {0x0};
	uint len = Packet::pack(data, &packet);
	Message message(data, len);
	mConnector->write(&message);

	// 设置已完成的数据长度
	mContext->setBytesLoaded(mProgress);

	// 通知监听器开始
	if (NULL != mListener)
	{
		mListener->expressProgress(mContext);
	}
}

} // end namespace cell
